package com.hrt.hudi.spark

import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
 * 往hudi里插入数据
 */
object InsertDataToHudi {
  def main(args: Array[String]): Unit = {
    val session: SparkSession = SparkSession.builder().master("local").appName("test")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // 必须使用这个格式
      .getOrCreate()

    session.sparkContext.setLogLevel("Error")

    val df: DataFrame = session.read.json("file:///F:\\P8\\workspace\\bigdatalake\\data\\jsondata.json")
    import org.apache.spark.sql.functions._
    // 增加一列分区字段 两个字段合并成一个分区
    val newdf: DataFrame = df.withColumn("partition_key",concat_ws("-",col("data_dt"),col("loc")))

    newdf.show()

    newdf.write.format("hudi")
      .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY,"id")  // 按照主键字段合并
      // 合并保留日期最新的一条，如果不指定该字段,则只保留按RECORDKEY_FIELD_OPT_KEY插入的第一条,设置了该字段才会按key合并最新的
      .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY,"data_dt")
//      .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY,"age")
      .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY,"partition_key") // 分区字段
      // 这两个并行度默认1500 太高了 调低
      .option("hoodie.upsert.shuffle.parallelism","2")
      .option("hoodie.insert.shuffle.parallelism","2")
      .option(HoodieWriteConfig.TABLE_NAME,"person_infos") // 创建的表名称，就是给spark的catalog保存的
      .mode(SaveMode.Overwrite)  // Overwrite是覆盖整个表的数据 所以建议用append
      .save("/hudi_data/person_infos") // 不能写hdfs://协议
  }
}
